#ifndef FILE_MANANGER_H
#define FILE_MANANGER_H

#include <muduo/base/Logging.h>
#include <muduo/base/Thread.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/InetAddress.h>


#include <sdfs/protobuf/Codec.h>
#include <sdfs/storage/Block.h>
#include <sdfs/storage/BlockStatus.h>
#include <sdfs/config/StorageConfig.h>
#include <sdfs/protobuf/Message.h>

#include <boost/shared_ptr.hpp>
#include <boost/weak_ptr.hpp>
#include <boost/any.hpp>

#include <google/protobuf/message.h>

#include <set>

using namespace muduo;
using namespace muduo::net;
using namespace boost;

using namespace sdfs::protobuf;

const int FileBufferSize = 1024*1024;

namespace sdfs
{
namespace storage
{

struct UploadContext
{
	string hashcode;
	PiecePtr piece;
	int req_id;
	bool needSync;
};

struct SendContext
{
	ReadOnlyFilePtr reader;
	PiecePtr piece;
};

typedef shared_ptr<UploadContext> UploadContextPtr;
typedef shared_ptr<SendContext> SendContextPtr;

struct SyncTask
{
	muduo::string hashcode;
	int timestamp;
};

typedef boost::shared_ptr<SyncTask> SyncTaskPtr;

class FileService
{
public:
	FileService(EventLoop *loop, 
			TcpConnectionPtr trackerConn, 
			ProtobufCodec* codec,
			int connectionSize): 
		loop_(loop),
		trackerConn_(trackerConn),
		statusTable_(new BlockStatusTable( 
		 	StorageConfig::storageDir(), 
		 	StorageConfig::blockNum(), 
		 	StorageConfig::blockSizeMB()*1024L*1024L,
		 	StorageConfig::blockReaderNum() ) ),
		codec_(codec),
		lastSyncTime_(-1),
		connSize_(connectionSize)
		
	{
		loop_->runEvery(StorageConfig::syncIntervalMs()/1000.0, 
			boost::bind(&FileService::syncTaskSchedual, this));
	}
	~FileService()
	{

	}

	void onFileSaveRequest(const TcpConnectionPtr& conn, 
		const shared_ptr<FileSaveRequestMessage>& message, 
		Timestamp time)
	{
		long size = message->filesize();
		PiecePtr piece = statusTable_->lockPiece(size);
		LOG_DEBUG << "save file request arrived";
		LOG_DEBUG << "conn reference: " << conn.use_count();
		FileSaveResponseMessage resp;
		resp.set_id(message->id()+1);
		resp.set_req_id(message->id());
		if(piece != NULL)
		{
			LOG_DEBUG << "lock piece : <"<< piece->getOffset() <<", "<<
				piece->getLength() << ">";
			resp.set_result(sdfs::protobuf::FileSaveResponseMessage::ACCEPT);
			shared_ptr<UploadContext> context(new UploadContext());
			context->piece = piece;
			context->hashcode = message->hashcode().c_str();
			context->req_id = message->id();
			context->needSync = true;
			conn->setContext(context);
			conn->setMessageCallback(
				boost::bind(&FileService::onSaveFile, this, _1, _2, _3));
			codec_->Send(conn, resp);

		}
		else
		{
			resp.set_result(sdfs::protobuf::FileSaveResponseMessage::REFUSE);
			resp.set_reason("No free block");
			codec_->Send(conn, resp);
			conn->shutdown();
		}
	}

	void onSaveFile(const TcpConnectionPtr& conn, Buffer* buffer, Timestamp time)
	{
		// LOG_DEBUG << "Saveing File";
		UploadContextPtr context = boost::any_cast<UploadContextPtr>(conn->getContext() );
		PiecePtr piece = context->piece;
		bool flag = statusTable_->SaveData( piece,  &(*buffer));
		if(flag == true)
		{
			LOG_DEBUG << "transport file END";
			FileSaveResultMessage resp;
			resp.set_id(msgIndex_.getAndAdd(1));
			resp.set_req_id(context->req_id);
			if(sendUpdateRequest(context))
			{
				resp.set_result(sdfs::protobuf::FileSaveResultMessage::SUCCESS);
			}
			else
			{
				resp.set_result(sdfs::protobuf::FileSaveResultMessage::FAILED);
			}
			codec_->Send(conn, resp);
			boost::any anyptr;
			conn->setContext(anyptr);
			if(context->needSync == false)
			{
				loop_->queueInLoop(boost::bind(&FileService::closeStorageConnction, this, conn));
				//closeStorageConnction(conn);
			}
			else
			{
				conn->shutdown();
			}
		}
		// else
		// {
		// 	LOG_DEBUG << "transport continues..."; 
		// }
	}

	

	void onAddSyncTask(const TcpConnectionPtr& conn, 
		const shared_ptr<FileSyncMessage>& message, 
		Timestamp time)
	{
		int size = message->hashcode_size();
		for (int i = 0; i < size; ++i)
		{
			appendTask(message->hashcode(i).c_str(), message->synctime());
			LOG_DEBUG << "add sync task: " << message->hashcode(i).c_str();
		}
		/*
		sdfs::StorageConfig::flush();
		
		RepeatedPtrField<muduo::string>* hashcodes = message->mutable_hashcode();
		RepeatedPtrField<muduo::string>::iterator it = hashcodes->begin();
		for ( ; it != hashcodes->end(); it++)
		{
			appendTask(*it);
		}
		*/
	}

	void onDownloadFile(const TcpConnectionPtr& conn, 
		const shared_ptr<FileDownloadRequestMessage>& message, 
		Timestamp time)
	{
		LOG_DEBUG << "start download: <"<< message->piece().blockid() << ", "
			<< message->piece().offset() << "," << message->piece().length()<< ">";
		PiecePtr piece(new Piece(message->piece().blockid(), 
			message->piece().offset(), message->piece().length()));
		ReadOnlyFilePtr reader = statusTable_->getReader(piece);
		FileDownloadResultMessage resp;
		resp.set_id(message->id()+1);
		resp.set_hashcode(message->hashcode());
		resp.set_req_id(message->id());
		if(reader != NULL)
		{
			LOG_DEBUG << "read piece : <"<< piece->getOffset() <<", "<<
				piece->getLength() << ">";
			resp.set_result(sdfs::protobuf::FileDownloadResultMessage::FOUND);
			resp.set_filesize(piece->getLength());
			SendContextPtr context(new SendContext());
			context->reader = reader;
			context->piece = piece;
			conn->setContext(context);
			codec_->Send(conn, resp);
		}
		else
		{
			resp.set_result(sdfs::protobuf::FileDownloadResultMessage::NOTEXIST);
			codec_->Send(conn, resp);
			conn->shutdown();
		}
	}  

	//send FindRequest to Tracker
	void syncTaskSchedual()
	{
		if( trackerConn_ != NULL && trackerConn_->connected())
		{
			if(restConnectionNum() <= 0)
			{
				return;
			}
			SyncTaskPtr task = getTask();
			if(task != NULL)
			{
				LOG_DEBUG << "look for: " <<task->hashcode.c_str();
				FileFindMessage request;
				request.set_id(msgIndex_.getAndAdd(1));
				request.set_hashcode(task->hashcode.c_str());
				sdfs::StorageConfig::updateLastSyncTime(task->timestamp);
				lastSyncTime_ = task->timestamp;
				codec_->Send(trackerConn_, request);
			}
			else
			{
				return;
			}
		}
	}
	// receive FindResult from tracker, connect target storage
	void onSyncFile(const TcpConnectionPtr& conn, 
		const shared_ptr<FileFindResultMessage>& message, 
		Timestamp time)
	{
		if(FileFindResultMessage::NOTEXIST == message->result())
		{
			// if not found, means the file has been deleted,
			// give up sync
			LOG_INFO << "file: " << message->hashcode() << "already removed, give up sync";
			return;
		}
		if(FileFindResultMessage::EXISTED == message->result())
		{
			// if already existed, means the file has been deleted,
			// give up sync
			LOG_INFO << "file: " << message->hashcode() << "already existed, give up sync";
			return;
		}
		// try to download file from given storage node
		// connect to ip:port, need a connection pool, which
		// can manage connection and help to prevent too many
		// connections run at a period, the connection pool
		// should be non-block. before send FileFindRequesetMessage
		// to tracker, FileService should try to find is there
		// enough connectors for the task. For the reason, 
		// onSyncFile will not care about connector.
		LOG_INFO << "file: " << message->hashcode() << "start sync";

		shared_ptr<FileDownloadRequestMessage> request(new FileDownloadRequestMessage());
		request->set_id(msgIndex_.getAndAdd(1));
		request->set_hashcode(message->hashcode());
		Filepiece* piece = request->mutable_piece();
		piece->set_blockid(message->piece().blockid());
		piece->set_offset(message->piece().offset());
		piece->set_length(message->piece().length());

		buildStorageConnection(
			message->worker().ip().c_str(), message->worker().port(), request);

	} 

	void storageConnectionCallback(const TcpConnectionPtr& conn, 
		shared_ptr<FileDownloadRequestMessage> message)
	{
		LOG_TRACE << "[" << conn->localAddress().toIpPort()
			<< "] -> [" << conn->peerAddress().toIpPort() << "]" <<
			" is " << (conn->connected() ? "UP" : "DOWN");
		// as long as stroage connection has been build,
		// first step is send FileDownloadRequestMessage
		codec_->Send(conn, *message);
	}

	void onRecieveFile(const TcpConnectionPtr& conn, 
		const shared_ptr<FileDownloadResultMessage>& message, 
		Timestamp time)
	{
		if(message->result() == sdfs::protobuf::FileDownloadResultMessage::NOTEXIST)
		{
			return;
		}
		long size = message->filesize();
		PiecePtr piece = statusTable_->lockPiece(size);
		LOG_DEBUG << "save file request arrived";
		FileReadyMessage resp;
		resp.set_id(message->id()+1);
		resp.set_req_id(message->id());
		if(piece != NULL)
		{
			shared_ptr<UploadContext> context(new UploadContext());
			context->piece = piece;
			context->hashcode = message->hashcode().c_str();
			context->req_id = message->id();
			context->needSync = false;
			conn->setContext(context);
			conn->setMessageCallback(
				boost::bind(&FileService::onSaveFile, this, _1, _2, _3));
			codec_->Send(conn, resp);

		}
		else
		{
			//fix me: should tell requestor "TOO BUSY!"
			conn->shutdown();
		}
	}

	void onSendFile(const TcpConnectionPtr& conn, 
		const shared_ptr<FileReadyMessage>& message, 
		Timestamp time)
	{
		LOG_DEBUG << "ready to send file";
		SendContextPtr context = boost::any_cast<SendContextPtr>(conn->getContext() );
		PiecePtr piece = context->piece;
		ReadOnlyFilePtr reader = context->reader;
		sendFile(conn, reader, piece);
		conn->shutdown();
	}

	int lastSyncTime()
	{
		return lastSyncTime_;
	}

	void setTrackerConnection(const TcpConnectionPtr& conn)
	{
		trackerConn_ = conn;
	}

	void resetTrackerConnection()
	{
		trackerConn_.reset();
	}

	void returnPiece(PiecePtr piece)
	{
		statusTable_->delData(piece);
	}
			

private:

	void sendFile(const TcpConnectionPtr& conn, ReadOnlyFilePtr reader, PiecePtr piece)
	{
		LOG_DEBUG << "start to send File";
		long nsend = 0;
		long n = 0;
		long nrest= piece->getLength();
		char buf[FileBufferSize];
		reader->seekFromBeginning(piece->getOffset());
		while(conn->connected() && nrest > 0)
		{
			if(nrest > FileBufferSize)
				n = FileBufferSize;
			else
				n = nrest;
			n = reader->read(buf, n);
			conn->send(buf, n);
			nsend += n;
			nrest -= n;
		}
		//assert(nsend == piece->getLength());
	}

	bool sendUpdateRequest(shared_ptr<UploadContext> context)
	{
		FileUpdateMessage message;
		message.set_id(msgIndex_.getAndAdd(1));
		message.set_hashcode(context->hashcode.c_str());
		Filepiece* piece = message.mutable_piece();
		piece->set_blockid(context->piece->getIndex());
		piece->set_offset(context->piece->getOffset());
		piece->set_length(context->piece->getLength());
		message.set_sync(context->needSync);
		message.set_groupid(sdfs::StorageConfig::groupIdx());
		message.set_storagename(sdfs::StorageConfig::name().c_str());
		if(trackerConn_ != NULL)
		{
			LOG_DEBUG << "update: " << context->hashcode;
			codec_->Send(trackerConn_, message);
			return true;
		}
		return false;
	}

	void appendTask(muduo::string hashcode, int time)
	{
		MutexLockGuard guard(syncMutex_);
		SyncTaskPtr task(new SyncTask());
		task->hashcode = hashcode;
		task->timestamp = time;
		syncQueue_.push_back(task);
	}

	SyncTaskPtr getTask()
	{
		MutexLockGuard guard(syncMutex_);
		std::list<SyncTaskPtr>::iterator it = 
			syncQueue_.begin();
		if(it != syncQueue_.end())
		{
			SyncTaskPtr task = *it;
			syncQueue_.erase(it);
			return task;
		}
		SyncTaskPtr task;
		return task;
	}

	int getConnectorSize()
	{
		MutexLockGuard guard(connMutex_);
		return connectors_.size();
	}

	shared_ptr<TcpClient> buildStorageConnection(muduo::string ip, int port,
		shared_ptr<FileDownloadRequestMessage> message)
	{
		MutexLockGuard guard(connMutex_);

		shared_ptr<TcpClient> client;
		LOG_DEBUG << "ip: " << ip << ", port: " << port;
		sockaddr_in sockaddr;
		sockaddr.sin_family = AF_INET;
		sockaddr.sin_port = htobe16(port);
		if (::inet_pton(AF_INET, ip.c_str(), &sockaddr.sin_addr) <= 0)
		{
		    LOG_SYSERR << "sockets::fromIpPort";
		    return client;
		}
		muduo::net::InetAddress addr(sockaddr);

		client.reset(
			new TcpClient(loop_, addr, "SyncConnection"));

		client->setConnectionCallback(
			boost::bind(&FileService::storageConnectionCallback, this, _1, message));
		client->setMessageCallback(
			boost::bind(&ProtobufCodec::onMessage, &(*codec_), _1, _2, _3));

		connectors_.insert(client);
		LOG_DEBUG << "client connect lunching, connectors_.size: " << connectors_.size();
		client->connect();
		
		return client;
	}

	void closeStorageConnction(const TcpConnectionPtr& conn)
	{
		MutexLockGuard guard(connMutex_);
		std::set< shared_ptr<TcpClient> >::iterator it = connectors_.begin();
		while(it != connectors_.end())
		{
			shared_ptr<TcpClient> client = *it;
			if(client->connection() == conn)
			{
				connectors_.erase(it);
				conn->connectDestroyed();
				client.reset();
				LOG_DEBUG << "disconnect, remove tcpclient" ;
				return;
			}
			it++;
		}
	}
	// not atomic function
	int restConnectionNum()
	{
		return connSize_ - connectors_.size();
	}

	/* data */
private:
	EventLoop* loop_;

	shared_ptr<TcpConnection> trackerConn_;
	BlockStatusTablePtr statusTable_;
	ProtobufCodec* codec_;
	AtomicInt32 msgIndex_;

	std::list<SyncTaskPtr> syncQueue_;
	MutexLock syncMutex_;
	int lastSyncTime_;

	std::set< shared_ptr<TcpClient> > connectors_;
	MutexLock connMutex_;
	int connSize_;
	
};


}
}
#endif